import os
import logging
from cryptography.fernet import Fernet
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import hmac
import hashlib
from confluent_kafka.schema_registry import SchemaRegistryClient

class SecureConfigLoader(FileSystemEventHandler):
    ENTERPRISE_DEFAULTS = {
        'bootstrap.servers': 'prod-kafka.acme.org:9093',
        'security.protocol': 'SASL_SSL',
        'sasl.mechanisms': 'SCRAM-SHA-512',
        'ssl.ciphers': 'TLS_AES_256_GCM_SHA384',
        'ssl.min_version': 'TLSv1.3'
    }

    def __init__(self, config_path, secret_key):
        self.config_path = config_path
        self.secret_key = secret_key.encode()
        self.config = self._load_config()
        self.observer = Observer()
        self.observer.schedule(self, path=os.path.dirname(config_path), recursive=False)
        self.observer.start()
        
        # Validate against schema registry
        self.schema_registry_client = SchemaRegistryClient({
            'url': self.get('schema.registry.url', 'https://schema.acme.org'),
            'ssl.ca.location': self.get('ssl.ca.location', '/etc/ssl/certs/ca-certificates.crt')
        })
        self._validate_schemas()

    def _load_config(self):
        """Load and decrypt configuration with HMAC validation"""
        try:
            with open(self.config_path, 'rb') as f:
                encrypted_data = f.read()
                
            # Split HMAC signature and payload
            received_signature, _, payload = encrypted_data.partition(b'|')
            computed_signature = hmac.new(self.secret_key, payload, hashlib.sha512).hexdigest()
            
            if not hmac.compare_digest(received_signature, computed_signature.encode()):
                raise ValueError("Invalid config signature")
                
            cipher = Fernet(self.secret_key)
            decrypted = cipher.decrypt(payload).decode()
            return {**self.ENTERPRISE_DEFAULTS, **eval(decrypted)}
            
        except Exception as e:
            logging.error(f"Config load failed: {str(e)}")
            return self.ENTERPRISE_DEFAULTS

    def _validate_schemas(self):
        """Validate compatibility with schema registry"""
        try:
            subjects = self.schema_registry_client.get_subjects()
            if 'kafka-config' not in subjects:
                raise ValueError("Missing schema registry entry")
        except Exception as e:
            logging.error(f"Schema validation failed: {str(e)}")

    def on_modified(self, event):
        """Hot-reload config on file change"""
        if event.src_path == self.config_path:
            logging.info("Reloading updated config")
            self.config = self._load_config()

    def get(self, key, default=None):
        """Get config value with environment override"""
        env_var = f"KAFKA_{key.upper().replace('.', '_')}"
        return os.getenv(env_var, self.config.get(key, default))

    def get_auditable_config(self):
        """Get config with sensitive fields masked"""
        sensitive_keys = {'sasl.password', 'ssl.key.password'}
        return {k: '*****' if k in sensitive_keys else v for k, v in self.config.items()}

if __name__ == '__main__':
    # Example initialization with encrypted config
    loader = SecureConfigLoader(
        config_path='/etc/app/kafka.conf.enc',
        secret_key=os.getenv('CONFIG_SECRET')
    )
    print(loader.get('bootstrap.servers'))